package com.bw.medical.realtime.common.base;


import com.bw.medical.realtime.common.constant.Constant;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public abstract class BaseSqlApp {

    //初始化
    public abstract void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableenv, String sql);

    public void start(String GroupId, int Parallelism, int FlinkPort) {
//        设置hadoop执行用户
        System.setProperty("HADOOP_USER_NAME", "root");
//        创建flink流环境
        Configuration configuration = new Configuration();
//        设置端口
        configuration.setInteger("rest.port", FlinkPort);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
//        设置并行度
        env.setParallelism(Parallelism);
////       设置CK
////        设置桩体后端为hashmap 适合本地开发
//        env.setStateBackend(new HashMapStateBackend());
////        五秒检查一次 设置检查频率 单位为毫秒
//        env.enableCheckpointing(5000);
////        设置检查模式  确保数据的准确性和唯一性 一般用在金融交易 和计费系统中
//        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
////        设置检查点存储状态节点路径
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop103:8020/medical/stream/" + GroupId);
////        设置最大并行度
//        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
////        设置两次检查之间的最小暂停时间
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(5000);
////        设置检查点超时时间
//        env.getCheckpointConfig().setCheckpointTimeout(10000);
//        设置当程序取消时是否保留检查点
//        RETAIN_ON_CANCELLATION：当作业被取消时，保留所有的外部化检查点。这意味着即使作业被取消，检查点数据也不会被删除。
//DELETE_ON_CANCELLATION：当作业被取消时，删除所有的外部化检查点。这是默认行为。
//NO_EXTERNALIZE_CHECKPOINTS：不外部化检查点。检查点只存储在 JobManager 中，如果 JobManager 挂掉，检查点也会丢失。
//        env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
//传递参数
        handle(env, tableEnv, GroupId);
//        执行
//        try {
//            env.execute(GroupId);
//        } catch (Exception e) {
//            e.printStackTrace();
//        }
    }
    public void createBaseDic(StreamTableEnvironment tableEnv){
        tableEnv.executeSql("CREATE TABLE base_dic (\n" +
                " rowkey STRING,\n" +
                " info ROW<dic_name STRING>,\n" +
                " PRIMARY KEY (rowkey) NOT ENFORCED\n" +
                ") WITH (\n" +
                " 'connector' = 'hbase-2.2',\n" +
                " 'table-name' = 'gmall:dim_base_dic',\n" +
                " 'zookeeper.quorum' = '"+ Constant.HBASE_ZOOKEEPER_QUORUM+"'\n" +
                ")");
    }
}
